import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.*;

class InputRPAByR implements WritableComparable {
	Long article_id;
	Long rating;
	
	InputRPAByR() {
		this.article_id = 0L;
		this.rating = 0L;
	}
	
	public String toString() {
		return (this.article_id.toString() + " " + this.rating.toString());
	}
	
	public void readFields(DataInput arg0) throws IOException {
		this.article_id = arg0.readLong();
		this.rating = arg0.readLong();
	}

	public void write(DataOutput arg0) throws IOException {
		arg0.writeLong(article_id);
		arg0.writeLong(rating);
	}

	public int compareTo(Object arg0) {
		InputRPAByR ip = (InputRPAByR) arg0;
		if (this.article_id < ip.article_id) 
			return -1;
		if (this.article_id > ip.article_id)
			return 1;
		if (this.rating < ip.rating)
			return -1;
		if (this.rating > ip.rating)
			return 1;	
		return 0;
	}
} 

public class Rating extends Configured implements Tool {
	
	public static class Map extends
			Mapper<LongWritable, Text, InputRPAByR, IntWritable> {
		
		private InputRPAByR input = new InputRPAByR();

		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			StringTokenizer tokenizer = new StringTokenizer(line);
			input.article_id = Long.parseLong((tokenizer.nextToken()));
			tokenizer.nextToken();
			input.rating = Long.parseLong((tokenizer.nextToken()));
			context.write(input, new IntWritable(1));
		}
	}

	public static class Reduce extends
			Reducer<InputRPAByR, IntWritable, InputRPAByR, IntWritable> {
		public void reduce(InputRPAByR key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			context.write(key, new IntWritable(sum));
		}
	}

	public int run(String[] args) throws Exception {
		/* 
		 * Calculate ratings per article
		 */
		Job job = new Job(getConf());
		job.setJarByClass(Rating.class);
		job.setJobName("Rating");

		job.setOutputKeyClass(InputRPAByR.class);
		job.setOutputValueClass(IntWritable.class);

		job.setMapperClass(Map.class);
		job.setCombinerClass(Reduce.class);
		job.setReducerClass(Reduce.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		boolean success = job.waitForCompletion(true);
		return success ? 0 : 1;
	}

	 
	 public static void main(String[] args) throws Exception {
		int ret = ToolRunner.run(new Rating(), args);
		System.exit(ret);
	}

}
